package com.ptkc.core.canal.handler;

import com.alibaba.otter.canal.protocol.CanalEntry;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.map.CaseInsensitiveMap;
import org.apache.commons.lang.StringUtils;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author: Mr.Jiao
 * @date: 2018年09月29日
 */
@Slf4j
public abstract class AbstractEventHandler implements EventHandler {

    private String database;

    private String table;

    public AbstractEventHandler(String database, String table){
        this.database = database;
        this.table = table;
    }

    @Override
    public final String getDatabase() {
        return database;
    }

    @Override
    public final String getTable() {
        return table;
    }

    @Override
    public void handle(CanalEntry.Entry entry, CanalEntry.RowChange change) {
        for(CanalEntry.RowData rowData : change.getRowDatasList()) {
            log.debug("handle change row data:{}", rowData.toString());
            handle(entry, rowData);
        }
    }

    protected String getAfterColumnValue(CanalEntry.RowData rowData, String columnName) {
        List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
        for(CanalEntry.Column column : columns) {
            if(column.getName().equals(columnName)) {
                return column.getValue();
            }
        }
        return null;
    }

    protected String getBeforeColumnValue(CanalEntry.RowData rowData, String columnName) {
        List<CanalEntry.Column> columns = rowData.getBeforeColumnsList();
        for(CanalEntry.Column column : columns) {
            if(column.getName().equals(columnName)) {
                return column.getValue();
            }
        }
        return null;
    }

    public Map<String, String> getRowMap(List<CanalEntry.Column> columns) {
        Map<String, String> row = new CaseInsensitiveMap();
        columns.forEach(column -> {
            String value = null;
            if(StringUtils.isNotEmpty(column.getValue())){
                value = column.getValue();
            }
            row.put(column.getName(), value);
        });
        return row;
    }

    public abstract void handle(CanalEntry.Entry entry, CanalEntry.RowData rowData);
}
